package com.toutiao.issue

import scala.math.sqrt

import org.apache.log4j.Logger

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}

class SimpleCollector(val minS: Double = 3.0) {
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
	private
	def norm(vs:Map[String,Double]) = {
	  if (vs.size > 0){
	    var s = vs.map(x => x._2*x._2).reduce(_+_)
	    s = sqrt(s)
	    vs.map(x => {
	      (x._1,x._2/s)
	    })
	  }
	  else {
	    vs
	  }
	}
  
  private
  val K = 24
  private
  val Index = new Array[InvertIndex[String]](K)
  for (j <- 0 until K){
    Index(j) = new InvertIndex[String]
  }
  
  def insertIndex(i:Long, vs:Array[String]) {
    val j = i%K
    Index(j.toInt).insert(i,vs.filter(_.size > 1)) // 不对单字建索引
  }
  
  def removeIndex(i:Long,vs:Array[String]) {
    val j = i%K
    Index(j.toInt).remove(i,vs.filter(_.size > 1))
  }
  
  var data = Map[String,Topic]()
  
  def size = data.size
  def apply(name:String) = data(name)
  def isIn(name:String) = data.contains(name)
  
  def keySet = data.keySet
  
  private
  def extract(topic:Topic):Array[String] = topic.vs.keySet.toArray
  
  def insertData(name:String, topic:Topic) {
    data += name->topic
  }
  
  private
  def initTopic(x:Document) = {
    //new Topic(x.name,Array(x.name),norm(x.vs),x.title,1.0,x.date,x.date)
    //new Topic(x.name,Array(x.name),x.vs,x.title,x.vs.size.toDouble)
    val t = new Topic
    t.name = x.name
    t.ms = Array(x.name)
    t.vs = norm(x.vs)
    t.title = x.title
    t.s = 1.0
    t.ct = x.date
    t.ut = x.date
    t
  }
  
  private
  def cos(m1:Map[String,Double], s1:Double, m2:Map[String,Double], s2:Double):Double = {
    var s = 0.0
    for ( (k,v) <- m1){
      s += v*m2.getOrElse(k,0.0)
    }
    s/(s1*s2)
  }
  
  // 1.修改权重
  // 给非score中的词默认权重
  // 对单字、标点符号给予不同的默认权重
  // 2.关键词匹配
  // 对《》或“”或【】或（）符号内的词，若不在对比中，相似度为0
  // 对时间词，若不在对比中，相似度为0
  
  val p = """(\[.*?\])|(\《.*?\》)|(\“.*?\”)|(\【.*?\】)|(\(.*?\))|(\".*?\")|(\（.*?\）)""".r // 括号词
  val u = """([0-9]+年)|([0-9]+月[0-9]+日)""".r // 时间词
  
  def getMatchScore(m1:Topic,m2:Topic):Double = {
	  val title1 = m1.title.reduce(_+_)
	  val title2 = m2.title.reduce(_+_)
	  
	  val mp1 = p.findAllIn(title1).toArray.map(v => {
	    v.slice(1, v.size-1)
	  }).filter(_.size < 8)
	  val mp2 = p.findAllIn(title2).toArray.map(v => {
	    v.slice(1,v.size-1)
	  }).filter(_.size < 8)
	  val mu1 = u.findAllIn(title1).toArray
	  val mu2 = u.findAllIn(title2).toArray
	  
	  val mm1 = mp1++mu1
	  val mm2 = mp2++mu2
	  
//	  if (mm1.size > 0){
//	    val line = mm1.reduce(_+" "+_)
//	    _log.info(title1+" "+line)
//	  }
	  
	  val s1 =
	    if (mm1.size == 0)
	      mm1.size
	    else
	      mm1.map(v => {  if (title2.contains(v)) 1 else 0 }).reduce(_+_)
	      
	  val s2 = 
	    if (mm2.size == 0)
	      mm2.size
	    else
	      mm2.map(v => { if (title1.contains(v)) 1 else 0}).reduce(_+_)
	  
	  
	  if (s1 == mm1.size && s2 == mm2.size)
		  1.0
	  else
	      0.0
  }
  
  
  def getSimScore(m1:Topic,m2:Topic):Double = {
    val s = getMatchScore(m1,m2)
    if (s > 0)
      cos(m1.vs,m1.s,m2.vs,m2.s)
    else
      s
  }
  
  private
  def merge(m1:Map[String,Double],n1:Int,m2:Map[String,Double],n2:Int) = {
    val n = n1+n2
    var z = m1
    for ( (k,v) <- m2){
      var x = v*n2+z.getOrElse(k,0.0)*n1
      z += k->x/n
    }
    var s = z.map(x => x._2*x._2).reduce(_+_)
    s = sqrt(s)
    z.map( x => {
      (x._1,x._2/s)
    })
  }
  
  def remove(name:String,i:Long) {
    if (data.contains(name)) {
      val vs = data(name).vs.map(_._1).toArray
      data -= name
      removeIndex(i,vs)
    }
  }
  
  private
  def retrieve(vs:Map[String,Double],S:Double) = {
    val es = (0 until K).map(j => {
      new WAND(Index(j),vs,S)
    })
    // !!!!! 
    val ns = es.par.flatMap(e => e.retrieve).seq
    ns
  }
  
  def process(
		  M1:HMap[String,Long],
		  M2:HMap[Long,String],
		  x:Document,
		  S:Double
      ) = {
    val event = initTopic(x)
    
    val ns = retrieve(
        event.title.filter(_.size > 1).map(v => { // 不对单字做召回
        	(v,1.0)
        	}).toMap, minS).filter(M2.contains(_)).map(M2(_)).filter(isIn(_))
    val rs = ns.par.map( n => {
      //(n,cos(data(n).vs, data(n).s, event.vs, event.s))
      (n,getSimScore(data(n),event))
    }).seq
    
    val (z,s) = if (rs.size > 0) rs.maxBy(_._2) else ("",0.0)
    //_log.info(z+" "+s)
    if (s >= S){ // Found Similar
      val eventD = data(z)
      val vs = eventD.vs
      val g = eventD.s
      val ms = eventD.ms ++ event.ms
      val title = eventD.title
      
      val ct = 
        if (eventD.ct.getTime < event.ct.getTime) eventD.ct
        else event.ct
      val ut =
        if (eventD.ut.getTime < event.ut.getTime) event.ut
        else eventD.ut
      
      data -= z
      val tp = new Topic
      tp.name = z
      tp.ms = ms
      tp.vs = vs
      tp.title = title
      tp.s = g
      tp.ct = ct
      tp.ut = ut
      //data += z->(new Topic(z,ms,vs,title,g,ct,ut))
      data += z->tp
      
      z
    }
    else {
      data += event.name->event
      val i = M1(event.name)
      insertIndex(i,event.title)
      
      event.name
    }
  }  
}

